package com.heima.kafka.simple;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class StreamTest {

    public static void main(String[] args) {


        Properties props = new Properties();
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,"kafka-stream-demo");


        StreamsBuilder sb = new StreamsBuilder();
        KStream<String, String> stream = sb.stream("topicA");//指明监听哪个主题（topic）,并获取对应的流


        stream
                .flatMapValues(new ValueMapper<String, Iterable<String>>() {
                    @Override
                    public Iterable<String> apply(String value) {
                        String[] split = value.split(" ");
                        return Arrays.asList(split);
                    }
                })

                .groupBy(new KeyValueMapper<String, String, String>() {

                    @Override
                    public String apply(String key, String value) {
                        return value;
                    }
                })
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .count()   //分组后统计每组的数量
                .toStream()  //将统计后的结果放到stream流中
                .map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {    //指定发到下一个topicB时它的格式
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, Long value) {
                        return new KeyValue<>(key.key().toString(),value+"");
                    }
                })
                .to("topicB");


        Topology topology = sb.build();

        KafkaStreams kafkaStreams = new KafkaStreams(topology,props);   //创建流处理对象
        kafkaStreams.start();  //开始工作
    }
}
